package com.atguigu.test.dacahngmainshitidierji;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class MySource{
    private volatile Boolean flag = true;
    private AtomicInteger atomicInteger = new AtomicInteger();
    private BlockingQueue<String> blockingQueue = null;
    //调用者传递具体的类型，通用接口参数写接口
    public MySource(BlockingQueue<String> blockingQueue){
        this.blockingQueue = blockingQueue;
        System.out.println("blockingQueue"+blockingQueue.getClass().getName());
    }
    //干活
    //生产
    public void product() throws InterruptedException {
        String data = null;
        Boolean offer = null;
        while (flag){
            data = atomicInteger.incrementAndGet()+"";
            offer = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
            if (offer){
                System.out.println(Thread.currentThread().getName()+"\t 插入成功");
            }else{
                System.out.println(Thread.currentThread().getName()+"\t 插入失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName()+"\t 打完收工了生产者");
    }
    //消费
    public void consumer() throws Exception{
        String offer = null;
        while (flag){
            offer = blockingQueue.poll( 2L, TimeUnit.SECONDS);

            if (null == offer || offer.equalsIgnoreCase("")){
                flag =false;
                System.out.println(Thread.currentThread().getName()+"\t 别几把消费了都他么没了");
                //必须退出要不然一直阻塞
                return;
            }
            System.out.println(Thread.currentThread().getName()+"\t 消费了"+ offer);
        }
    }

    public void stop(){
        this.flag = false;
    }
}
/**
 * @Author: 19shusheng.com
 * @Description:
 * @Date: Created in 22:27 2020/11/12
 * @Modified By:    volatile+cas+atomicinteger,blockingwueue+线程交互
 */
public class ProductConsumer_BlockingQueueDemo {
    public static void main(String[] args) {
        MySource mySource = new MySource(new ArrayBlockingQueue<>(10));
        //生产
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+"\t生产线程启动");
                mySource.product();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"生产").start();
        //消费
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+"\t消费线程启动");
                mySource.consumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"消费").start();
        //让你们工作5秒
        try {
            TimeUnit.SECONDS.sleep(5);
            mySource.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
